package com.ctw.mapmycity.hbase.repository;

import java.util.List;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.data.hadoop.hbase.TableCallback;
import org.springframework.stereotype.Repository;

import com.ctw.mapmycity.model.Account;

@Repository
public class AccountRepository {
	
	@Autowired
	private HbaseTemplate hbaseTemplate;

	private String tableName = "account";
	public static byte[] CF_INFO = Bytes.toBytes("cfInfo");

	private byte[] qEmail = Bytes.toBytes("email");
	private byte[] qFullName = Bytes.toBytes("fullName");
	private byte[] qPassword = Bytes.toBytes("password");
	private byte[] qToken = Bytes.toBytes("token");
	private byte[] qAuthType = Bytes.toBytes("authType");
	
	public List<Account> findAll() {
		return hbaseTemplate.find(tableName, "cfInfo", new RowMapper<Account>() {
			@Override
			public Account mapRow(Result result, int rowNum) throws Exception {
				return new Account(Bytes.toString(result.getValue(CF_INFO, qEmail)),
									Bytes.toString(result.getValue(CF_INFO, qFullName)),
							    	Bytes.toString(result.getValue(CF_INFO, qPassword)),
							    	Bytes.toString(result.getValue(CF_INFO, qToken)),
							    	Bytes.toString(result.getValue(CF_INFO, qAuthType)));
			}
		});

	}

	public Account get(String rowName) {
		return hbaseTemplate.get(tableName, rowName, new RowMapper<Account>() {
			@Override
			public Account mapRow(Result result, int rowNum) throws Exception {
				return new Account(Bytes.toString(result.getValue(CF_INFO, qEmail)),
									Bytes.toString(result.getValue(CF_INFO, qFullName)),
							    	Bytes.toString(result.getValue(CF_INFO, qPassword)),
							    	Bytes.toString(result.getValue(CF_INFO, qToken)),
							    	Bytes.toString(result.getValue(CF_INFO, qAuthType)));
			}
		});

	}
	
	public Account save(final Account acc) {
		return hbaseTemplate.execute(tableName, new TableCallback<Account>() {
			public Account doInTable(HTableInterface table) throws Throwable {
				Put p = new Put(Bytes.toBytes(acc.getEmail()));
				p.add(CF_INFO, qEmail, Bytes.toBytes(acc.getEmail()));
				p.add(CF_INFO, qFullName, Bytes.toBytes(acc.getFullName()));
				p.add(CF_INFO, qPassword, Bytes.toBytes(acc.getPassword()));
				p.add(CF_INFO, qToken, Bytes.toBytes(acc.getToken()));
				p.add(CF_INFO, qAuthType, Bytes.toBytes(acc.getAuthType()));
				table.put(p);
				
				return acc;
			}
		});
	}
	
	public void delete (final String email) {
		hbaseTemplate.execute(tableName, new TableCallback<Object>() {
            public Object doInTable(HTableInterface hti) throws Throwable {
            	Delete deleteOp = new Delete(Bytes.toBytes(email));
            	hti.delete(deleteOp);
                return null;
            }
        });
	}

}
